{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 动量与反转策略分析\n",
    "\n",
    "本笔记本实现动量效应和反转效应的量化分析"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import akshare as ak\n",
    "import matplotlib.pyplot as plt\n",
    "from tqdm import tqdm\n",
    "\n",
    "# 设置中文显示\n",
    "plt.rcParams['font.sans-serif'] = ['SimHei']\n",
    "plt.rcParams['axes.unicode_minus'] = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据准备"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_stock_pool():\n",
    "    \"\"\"\n",
    "    获取股票池数据\n",
    "    \n",
    "    Returns:\n",
    "        list: 股票代码列表\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 获取沪深300成分股\n",
    "        df = ak.index_stock_cons_csindex(symbol=\"000300\")\n",
    "        return df['成分券代码'].tolist()\n",
    "    except Exception as e:\n",
    "        print(f\"获取股票池失败: {e}\")\n",
    "        return []\n",
    "\n",
    "def get_stock_data(stock_code, start_date='20100101', end_date='20231231'):\n",
    "    \"\"\"\n",
    "    获取单只股票历史数据\n",
    "    \n",
    "    Args:\n",
    "        stock_code: 股票代码\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 复权收盘价\n",
    "    \"\"\"\n",
    "    try:\n",
    "        df = ak.stock_zh_a_hist(symbol=stock_code, period=\"daily\", \n",
    "                              start_date=start_date, end_date=end_date, adjust=\"qfq\")\n",
    "        df['日期'] = pd.to_datetime(df['日期'])\n",
    "        return df.set_index('日期')['收盘']\n",
    "    except Exception as e:\n",
    "        print(f\"获取股票{stock_code}数据失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 动量策略实现"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def momentum_strategy(price_data, formation_period=20, holding_period=5, top_n=10):\n",
    "    \"\"\"\n",
    "    动量策略实现\n",
    "    \n",
    "    Args:\n",
    "        price_data: 各股票价格DataFrame\n",
    "        formation_period: 形成期(天)\n",
    "        holding_period: 持有期(天)\n",
    "        top_n: 选取前N只股票\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 策略收益率\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 计算形成期收益率\n",
    "        returns = price_data.pct_change(formation_period)\n",
    "        \n",
    "        # 初始化策略收益率\n",
    "        strategy_returns = pd.Series(index=price_data.index, dtype=float)\n",
    "        \n",
    "        # 滚动计算\n",
    "        for i in range(formation_period, len(price_data)-holding_period):\n",
    "            # 按形成期收益率排序\n",
    "            ranked = returns.iloc[i].sort_values(ascending=False)\n",
    "            \n",
    "            # 选取前top_n只股票\n",
    "            selected = ranked.head(top_n).index\n",
    "            \n",
    "            # 计算持有期收益率(等权重)\n",
    "            holding_return = price_data[selected].iloc[i+1:i+1+holding_period].pct_change().mean(axis=1)\n",
    "            \n",
    "            # 保存结果\n",
    "            strategy_returns.iloc[i+1:i+1+holding_period] = holding_return.values\n",
    "            \n",
    "        return strategy_returns.dropna()\n",
    "    except Exception as e:\n",
    "        print(f\"动量策略执行失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 反转策略实现"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def reversal_strategy(price_data, formation_period=20, holding_period=5, top_n=10):\n",
    "    \"\"\"\n",
    "    反转策略实现\n",
    "    \n",
    "    Args:\n",
    "        price_data: 各股票价格DataFrame\n",
    "        formation_period: 形成期(天)\n",
    "        holding_period: 持有期(天)\n",
    "        top_n: 选取前N只股票\n",
    "        \n",
    "    Returns:\n",
    "        pd.Series: 策略收益率\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # 计算形成期收益率\n",
    "        returns = price_data.pct_change(formation_period)\n",
    "        \n",
    "        # 初始化策略收益率\n",
    "        strategy_returns = pd.Series(index=price_data.index, dtype=float)\n",
    "        \n",
    "        # 滚动计算\n",
    "        for i in range(formation_period, len(price_data)-holding_period):\n",
    "            # 按形成期收益率排序(升序)\n",
    "            ranked = returns.iloc[i].sort_values()\n",
    "            \n",
    "            # 选取前top_n只股票(表现最差)\n",
    "            selected = ranked.head(top_n).index\n",
    "            \n",
    "            # 计算持有期收益率(等权重)\n",
    "            holding_return = price_data[selected].iloc[i+1:i+1+holding_period].pct_change().mean(axis=1)\n",
    "            \n",
    "            # 保存结果\n",
    "            strategy_returns.iloc[i+1:i+1+holding_period] = holding_return.values\n",
    "            \n",
    "        return strategy_returns.dropna()\n",
    "    except Exception as e:\n",
    "        print(f\"反转策略执行失败: {e}\")\n",
    "        return pd.Series()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 主分析流程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 获取股票池\n",
    "stock_pool = get_stock_pool()[:50]  # 取前50只股票减少计算量\n",
    "\n",
    "# 获取各股票数据\n",
    "price_data = pd.DataFrame()\n",
    "for code in tqdm(stock_pool, desc='获取股票数据'):\n",
    "    price_data[code] = get_stock_data(code)\n",
    "\n",
    "# 运行动量策略\n",
    "momentum_returns = momentum_strategy(price_data)\n",
    "\n",
    "# 运行反转策略\n",
    "reversal_returns = reversal_strategy(price_data)\n",
    "\n",
    "# 计算累计收益\n",
    "cum_momentum = (1 + momentum_returns).cumprod()\n",
    "cum_reversal = (1 + reversal_returns).cumprod()\n",
    "\n",
    "# 绘制结果\n",
    "plt.figure(figsize=(12, 6))\n",
    "cum_momentum.plot(label='动量策略')\n",
    "cum_reversal.plot(label='反转策略')\n",
    "plt.title('动量与反转策略表现对比')\n",
    "plt.ylabel('累计收益')\n",
    "plt.legend()\n",
    "plt.grid()\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 策略评价指标"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def evaluate_strategy(returns, rf=0.03/252):\n",
    "    \"\"\"\n",
    "    计算策略评价指标\n",
    "    \n",
    "    Args:\n",
    "        returns: 策略收益率序列\n",
    "        rf: 无风险利率(日)\n",
    "        \n",
    "    Returns:\n",
    "        dict: 包含各项指标\n",
    "    \"\"\"\n",
    "    if returns.empty:\n",
    "        return {}\n",
    "        \n",
    "    try:\n",
    "        # 年化收益率\n",
    "        annual_return = (1 + returns.mean())**252 - 1\n",
    "        \n",
    "        # 年化波动率\n",
    "        annual_vol = returns.std() * np.sqrt(252)\n",
    "        \n",
    "        # 夏普比率\n",
    "        sharpe = (returns.mean() - rf) / returns.std() * np.sqrt(252)\n",
    "        \n",
    "        # 最大回撤\n",
    "        cum_returns = (1 + returns).cumprod()\n",
    "        peak = cum_returns.cummax()\n",
    "        drawdown = (cum_returns - peak) / peak\n",
    "        max_drawdown = drawdown.min()\n",
    "        \n",
    "        return {\n",
    "            '年化收益率': annual_return,\n",
    "            '年化波动率': annual_vol,\n",
    "            '夏普比率': sharpe,\n",
    "            '最大回撤': max_drawdown\n",
    "        }\n",
    "    except Exception as e:\n",
    "        print(f\"策略评价失败: {e}\")\n",
    "        return {}\n",
    "\n",
    "# 评价动量策略\n",
    "momentum_metrics = evaluate_strategy(momentum_returns)\n",
    "print(\"动量策略表现:\")\n",
    "for k, v in momentum_metrics.items():\n",
    "    print(f\"{k}: {v:.4f}\")\n",
    "\n",
    "# 评价反转策略\n",
    "reversal_metrics = evaluate_strategy(reversal_returns)\n",
    "print(\"\\n反转策略表现:\")\n",
    "for k, v in reversal_metrics.items():\n",
    "    print(f\"{k}: {v:.4f}\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}